package com.atguigu.flink.state;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

/**
 * Created by Smexy on 2023/4/11
 *
 *  StateBackend: 状态后端。
 *      作用： 管理(读，写，备份)状态
 *
 *      有两种实现:
 *          HashMapStateBackEnd： 默认。状态以HashMap存储在TM的堆内存中。
 *                  弊端： 状态是不能无限存储。大状态可能存不下。
 *                  优势： 读写效率高
 *
 *
 *          RockDBStateBackEnd: 手动设置，需要额外安装。 状态存储在一个内嵌式数据库 RockDB中。生产可用。
 *                  优势： 存储空间可以理解为无限，可以存储大状态
 *                  弊端:  效率低。
 *                              1.数据会先存入 写缓存(内存)，但是缓存满了后，会刷写到磁盘。
 *                              2.所有的数据，必须序列化为byte[]才能存入。
 *                                  读写状态时，需要额外的序列化和反序列化的开销
 *                        比HashMapStateBackEnd慢10倍。
 *
 *                 内嵌式数据库： 类似hive中的derby。数据库无需额外安装，启动一个JVM进程，自动创建一个数据库。
 *                                  jvm进程停止，数据库进程也停止。
 *
 *----------------------------
 *  通常状态后端不会在代码中写死，而是在配置中指定:
 *          state.backend: hashmap|rocksdb
 *
 *  ----------------------------------
 *      checkpoint:   指定状态备份到哪里。
 *          Checkpoint Storage
 *              默认:JobManagerCheckpointStorage,状态会定期备份到JM的内存中。 不可靠，但是效率高。
 *
 *              生产可用： 备份到一个文件系统。使用 FileSystemCheckpointStorage
 *
 *        配置文件: state.checkpoints.dir hdfs://xxxx
 *
 *
 *--------------------------
 *  ClassNotFoundException: org.apache.flink.state.changelog.ChangelogStateBackend
 *      在pom中添加: flink-statebackend-changelog
 */
public class Demo12_StateBackend
{
    public static void main(String[] args) {


        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        //是null，代表你自己没有明确设置，此时flink会给你进行默认设置
        System.out.println(env.getStateBackend());

        //代码中设置hashmap状态后端使用增量快照
        env.enableChangelogStateBackend(true);

        env.setParallelism(2);

        //切换使用rockdb状态后端
        //EmbeddedRocksDBStateBackend embeddedRocksDBStateBackend = new EmbeddedRocksDBStateBackend();
        //指定rocksdb使用增量快照
       // EmbeddedRocksDBStateBackend embeddedRocksDBStateBackend = new EmbeddedRocksDBStateBackend(true);
        //env.setStateBackend(embeddedRocksDBStateBackend);


        //修改ck的备份路径
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
       // checkpointConfig.setCheckpointStorage("file:///d:/tmp");
        checkpointConfig.setCheckpointStorage("hdfs://hadoop102:8020/ck");

        env.enableCheckpointing(2000);

        env
            .socketTextStream("hadoop102", 8888)
            .map(new Demo1_State.MyMapFunction())
            .addSink(new SinkFunction<String>()
            {
                @Override
                public void invoke(String value, SinkFunction.Context context) throws Exception {
                    if (value.contains("x")){
                        //模拟抛异常
                        throw new RuntimeException("出异常了...");
                    }
                    System.out.println(value);
                }
            });


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    //进行对每次输入的字符串进行累加拼接
    public static class MyMapFunction implements MapFunction<String,String>, CheckpointedFunction
    {


        private ListState<String> result;

        @Override
        public String map(String value) throws Exception {
            result.add(value);
            return result.get().toString();
        }


        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
        }

        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {

            OperatorStateStore operatorStateStore = context.getOperatorStateStore();

            ListStateDescriptor<String> listStateDescriptor = new ListStateDescriptor<>("strs", String.class);
            result = operatorStateStore.getListState(listStateDescriptor);
        }
    }
}
